/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.endpoint;

import java.util.*;
import org.apache.skywalking.apm.network.common.KeyStringValuePair;
import org.apache.skywalking.apm.network.language.agent.*;
import org.apache.skywalking.oap.server.core.*;
import org.apache.skywalking.oap.server.core.cache.*;
import org.apache.skywalking.oap.server.core.source.*;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
import org.apache.skywalking.oap.server.receiver.trace.provider.*;
import org.apache.skywalking.oap.server.receiver.trace.provider.parser.SpanTags;
import org.apache.skywalking.oap.server.receiver.trace.provider.parser.decorator.*;
import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.*;
import org.slf4j.*;

import static java.util.Objects.nonNull;

/**
 * Notice, in here, there are following concepts match
 *
 * v5        |   v6
 *
 * 1. Application == Service 2. Server == Service Instance 3. Service == Endpoint
 *
 * @author peng-yongsheng, wusheng
 */
public class MultiScopesSpanListener implements EntrySpanListener, ExitSpanListener, GlobalTraceIdsListener {

    private static final Logger logger = LoggerFactory.getLogger(MultiScopesSpanListener.class);

    private final SourceReceiver sourceReceiver;
    private final ServiceInstanceInventoryCache instanceInventoryCache;
    private final ServiceInventoryCache serviceInventoryCache;
    private final EndpointInventoryCache endpointInventoryCache;

    private final List<SourceBuilder> entrySourceBuilders;
    private final List<SourceBuilder> exitSourceBuilders;
    private final List<DatabaseSlowStatement> slowDatabaseAccesses;
    private final TraceServiceModuleConfig config;
    private SpanDecorator entrySpanDecorator;
    private long minuteTimeBucket;
    private String traceId;

    private MultiScopesSpanListener(ModuleManager moduleManager, TraceServiceModuleConfig config) {
        this.sourceReceiver = moduleManager.find(CoreModule.NAME).provider().getService(SourceReceiver.class);
        this.entrySourceBuilders = new LinkedList<>();
        this.exitSourceBuilders = new LinkedList<>();
        this.slowDatabaseAccesses = new ArrayList<>(10);
        this.instanceInventoryCache = moduleManager.find(CoreModule.NAME).provider().getService(ServiceInstanceInventoryCache.class);
        this.serviceInventoryCache = moduleManager.find(CoreModule.NAME).provider().getService(ServiceInventoryCache.class);
        this.endpointInventoryCache = moduleManager.find(CoreModule.NAME).provider().getService(EndpointInventoryCache.class);
        this.config = config;
        this.traceId = null;
    }

    @Override public boolean containsPoint(Point point) {
        return Point.Entry.equals(point) || Point.Exit.equals(point) || Point.TraceIds.equals(point);
    }

    /**
     * 在 EntrySpan 的 refs 字段中记录的一个（或多个） TraceSegmentReference ，分别指向了上一个（或是多个）上游系统的 TraceSegment，
     * SkyWalking OAP 可以通过 TraceSegmentReference 将两个系统的调用关系串联起来。
     *
     * 方法的核心就是来解析 EntrySpan.refs 字段中记录的全部 TraceSegmentReference 对象，为每一个 TraceSegmentReference 生成一个相
     * 应的 SourceBuilder 对象并记录下来（ entrySourceBuilders 字段，List 类型）。在 build() 方法中还会继续处理这些 SourceBuilder
     * 对象。
     */
    @Override
    public void parseEntry(SpanDecorator spanDecorator, SegmentCoreInfo segmentCoreInfo) {
        // 记录 TraceSegment 起始的分钟级时间窗口
        this.minuteTimeBucket = segmentCoreInfo.getMinuteTimeBucket();

        if (spanDecorator.getRefsCount() > 0) {
            for (int i = 0; i < spanDecorator.getRefsCount(); i++) {
                // 获取 TraceSegmentReference 对应的 ReferenceDecorator
                ReferenceDecorator reference = spanDecorator.getRefs(i);
                // 创建对应的SourceBuilder
                SourceBuilder sourceBuilder = new SourceBuilder();
                // 记录上游系统的 serviceId、serviceInstanceId 以及 endpointId
                sourceBuilder.setSourceEndpointId(reference.getParentEndpointId());

                if (spanDecorator.getSpanLayer().equals(SpanLayer.MQ)) { // 针对MQ组件的特殊处理
                    int serviceIdByPeerId = serviceInventoryCache.getServiceId(reference.getNetworkAddressId());
                    int instanceIdByPeerId = instanceInventoryCache.getServiceInstanceId(serviceIdByPeerId, reference.getNetworkAddressId());
                    sourceBuilder.setSourceServiceInstanceId(instanceIdByPeerId);
                    sourceBuilder.setSourceServiceId(serviceIdByPeerId);
                } else {
                    sourceBuilder.setSourceServiceInstanceId(reference.getParentServiceInstanceId());
                    sourceBuilder.setSourceServiceId(instanceInventoryCache.get(reference.getParentServiceInstanceId()).getServiceId());
                }
                // 记录下游(当前)系统的 serviceId、serviceInstanceId 以及 endpointId
                sourceBuilder.setDestEndpointId(spanDecorator.getOperationNameId());
                sourceBuilder.setDestServiceInstanceId(segmentCoreInfo.getServiceInstanceId());
                sourceBuilder.setDestServiceId(segmentCoreInfo.getServiceId());
                sourceBuilder.setDetectPoint(DetectPoint.SERVER);
                // 记录当前组件的类型
                sourceBuilder.setComponentId(spanDecorator.getComponentId());
                // 这里将解析 EntrySpan 和 ExitSpan 都会设置的一些公共信息封装到了setPublicAttrs中
                setPublicAttrs(sourceBuilder, spanDecorator);
                // 将 SourceBuilder 记录到 entrySourceBuilders 集合中
                entrySourceBuilders.add(sourceBuilder);
            }
        } else {
            SourceBuilder sourceBuilder = new SourceBuilder();
            sourceBuilder.setSourceEndpointId(Const.USER_ENDPOINT_ID);
            sourceBuilder.setSourceServiceInstanceId(Const.USER_INSTANCE_ID);
            sourceBuilder.setSourceServiceId(Const.USER_SERVICE_ID);
            sourceBuilder.setDestEndpointId(spanDecorator.getOperationNameId());
            sourceBuilder.setDestServiceInstanceId(segmentCoreInfo.getServiceInstanceId());
            sourceBuilder.setDestServiceId(segmentCoreInfo.getServiceId());
            sourceBuilder.setDetectPoint(DetectPoint.SERVER);
            sourceBuilder.setComponentId(spanDecorator.getComponentId());

            setPublicAttrs(sourceBuilder, spanDecorator);
            entrySourceBuilders.add(sourceBuilder);
        }
        this.entrySpanDecorator = spanDecorator;
    }

    // 将 SourceBuilder 记录到 entrySourceBuilders 集合中
    @Override public void parseExit(SpanDecorator spanDecorator, SegmentCoreInfo segmentCoreInfo) {
        if (this.minuteTimeBucket == 0) {
            this.minuteTimeBucket = segmentCoreInfo.getMinuteTimeBucket();
        }

        // 每个 ExitSpan 创建相应的 SourceBuilder 对象
        SourceBuilder sourceBuilder = new SourceBuilder();

        int peerId = spanDecorator.getPeerId();
        if (peerId == 0) {
            return;
        }
        // 在 Agent 创建 ExitSpan 的时候，只知道下游系统的 peer 信息（remotePeer 或 peerId），在 parseExit()
        // 方法中处理 ExitSpan 的时候，会通过下面这段代码将 peerId 转换成下游系统的 ServiceId：

        // 获取 peerId 在 service_inventory 中对应的 id
        int destServiceId = serviceInventoryCache.getServiceId(peerId);
        // 在 ServiceInventory 中有一个 mappingServiceId 字段，记录了 peerId 关联的真的 Service 的 Id。
        // 将 peerId 与对应 Service 的 serviceId 进行关联的逻辑是在 ServiceMappingSpanListener 中实现的。
        int mappingServiceId = serviceInventoryCache.get(destServiceId).getMappingServiceId();
        int destInstanceId = instanceInventoryCache.getServiceInstanceId(destServiceId, peerId);

        // 记录当前(上游)系统的 serviceId、serviceInstanceId 以及 endpointId
        sourceBuilder.setSourceEndpointId(Const.USER_ENDPOINT_ID);
        sourceBuilder.setSourceServiceInstanceId(segmentCoreInfo.getServiceInstanceId());
        sourceBuilder.setSourceServiceId(segmentCoreInfo.getServiceId());

        // 记录下游系统的 serviceId、serviceInstanceId以及endpointId
        sourceBuilder.setDestEndpointId(spanDecorator.getOperationNameId());
        sourceBuilder.setDestServiceInstanceId(destInstanceId);
        // 如果存在 mappingServiceId，则更新下游系统的 serviceId
        if (Const.NONE == mappingServiceId) {
            sourceBuilder.setDestServiceId(destServiceId);
        } else {
            sourceBuilder.setDestServiceId(mappingServiceId);
        }
        // Client 角色
        sourceBuilder.setDetectPoint(DetectPoint.CLIENT);
        sourceBuilder.setComponentId(spanDecorator.getComponentId());
        // 记录这些 id 对应的字符串
        setPublicAttrs(sourceBuilder, spanDecorator);
        // 将该 SourceBuilder 对象填充到 exitSourceBuilders 集合中，等待在 build() 方法中处理
        exitSourceBuilders.add(sourceBuilder);

        // 对 DB 慢查询的特殊处理：主要用于统计慢查询。
        // 这里的慢查询统计不仅是 DB 的慢查询，还包括其他常见的存储，例如：Redis、MongoDB 等等。
        if (sourceBuilder.getType().equals(RequestType.DATABASE)) {
            boolean isSlowDBAccess = false;

            DatabaseSlowStatement statement = new DatabaseSlowStatement();
            // 由 TraceSegment.id 以及 Span.id 构成的唯一标识
            statement.setId(segmentCoreInfo.getSegmentId() + "-" + spanDecorator.getSpanId());
            // 记录存储对应的 ServiceId
            statement.setDatabaseServiceId(sourceBuilder.getDestServiceId());
            // 此次慢查询的实际耗时
            statement.setLatency(sourceBuilder.getLatency());
            // 秒级时间窗口
            statement.setTimeBucket(TimeBucket.getSecondTimeBucket(segmentCoreInfo.getStartTime()));
            // 记录发生此次慢查询的 traceId
            statement.setTraceId(traceId);
            // 遍历 ExitSpan 携带的 Tag 信息
            for (KeyStringValuePair tag : spanDecorator.getAllTags()) {
                if (SpanTags.DB_STATEMENT.equals(tag.getKey())) {
                    // 具体执行的操作，例如，访问 DB 的话，就是 SQL 语句
                    statement.setStatement(tag.getValue());
                } else if (SpanTags.DB_TYPE.equals(tag.getKey())) {
                    // 在 application.yml 配置文件中配置了不同存储的慢查询阈值上限，
                    // 这里会根据 dbType（其值可以为 sql、Redis、MongoDB 等）查找其阈值
                    String dbType = tag.getValue();
                    DBLatencyThresholdsAndWatcher thresholds = config.getDbLatencyThresholdsAndWatcher();
                    int threshold = thresholds.getThreshold(dbType);
                    // 判断此次请求存储的操作是否为慢查询
                    if (sourceBuilder.getLatency() > threshold) {
                        isSlowDBAccess = true;
                    }
                }
            }

            // 将慢查询记录到 slowDatabaseAccesses 集合中
            if (isSlowDBAccess) {
                slowDatabaseAccesses.add(statement);
            }
        }
    }

    // 记录 id 对应的字符串
    private void setPublicAttrs(SourceBuilder sourceBuilder, SpanDecorator spanDecorator) {
        // 计算 latency，即当前 Span 的时间跨度，在后面计算系统延迟的时候会用到
        long latency = spanDecorator.getEndTime() - spanDecorator.getStartTime();
        sourceBuilder.setLatency((int)latency);
        sourceBuilder.setResponseCode(Const.NONE);
        sourceBuilder.setStatus(!spanDecorator.getIsError());

        switch (spanDecorator.getSpanLayer()) {
            case Http:
                sourceBuilder.setType(RequestType.HTTP);
                break;
            case Database:
                sourceBuilder.setType(RequestType.DATABASE);
                break;
            default:
                sourceBuilder.setType(RequestType.RPC);
                break;
        }

        // 填充上下游系统中的 serviceName、serviceInstanceName 以及 endpointName
        sourceBuilder.setSourceServiceName(serviceInventoryCache.get(sourceBuilder.getSourceServiceId()).getName());
        sourceBuilder.setSourceServiceInstanceName(instanceInventoryCache.get(sourceBuilder.getSourceServiceInstanceId()).getName());
        sourceBuilder.setSourceEndpointName(endpointInventoryCache.get(sourceBuilder.getSourceEndpointId()).getName());
        sourceBuilder.setDestServiceName(serviceInventoryCache.get(sourceBuilder.getDestServiceId()).getName());
        sourceBuilder.setDestServiceInstanceName(instanceInventoryCache.get(sourceBuilder.getDestServiceInstanceId()).getName());
        sourceBuilder.setDestEndpointName(endpointInventoryCache.get(sourceBuilder.getDestEndpointId()).getName());
    }

    // 方法主要负责将 entrySourceBuilders 和 exitSourceBuilders 两个集合中的全部 SourceBuilder
    // 转换成相应的 Source 对象，然后交给 SourceReceiver 中的 MetricsStreamProcessor 处理。
    @Override public void build() {
        entrySourceBuilders.forEach(entrySourceBuilder -> {
            entrySourceBuilder.setTimeBucket(minuteTimeBucket);
            // entrySourceBuilders 集合中的每个 SourceBuilder 对象都会生成多个 Source 对象

            // AllDispatcher 会将一个 All 对象转换成多个 AllP*Metrics 对象
            sourceReceiver.receive(entrySourceBuilder.toAll());
            // 集合中的每个 SourceBuilder 元素还会生成 一个 Service 对象
            sourceReceiver.receive(entrySourceBuilder.toService());
            sourceReceiver.receive(entrySourceBuilder.toServiceInstance());
            sourceReceiver.receive(entrySourceBuilder.toEndpoint());

            // 生成三个 Relation 类型的 Source 对象，Relation 表示的某种调用关系：
            // 例如：一个 ServiceRelation 对象记录了某两个 Service 之间的调用关系，
            //      一个 ServiceInstanceRelation 对象记录了某两个 ServiceInstance 之间的调用关系，
            //      一个 EndpointRelation 对象记录了某两个 Endpoint 之间的调用关系。
            // 指标生成的位置分成了 Server 端和 Client 端两类：
            // 其中，Server 类型指标产生的位置对应被调用系统（接收请求者）的 EntrySpan，而 Client 类型指标产生位置对应调用系统（发起请求者）的 ExitSpan。
            // 所以，在 entrySourceBuilders 集合中的 SourceBuilder 生成的都是 Server 端的 Relaiton 指标，而无法生成 Client 端的 Relation 指标，因为
            // entrySourceBuilders 集合是 EntrySpan 的处理结果。
            // 同理可知 exitSourceBuilders 集合，只会生成 Client 端的 Relation 指标。
            sourceReceiver.receive(entrySourceBuilder.toServiceRelation());
            sourceReceiver.receive(entrySourceBuilder.toServiceInstanceRelation());
            EndpointRelation endpointRelation = entrySourceBuilder.toEndpointRelation();
            /**
             * Parent endpoint could be none, because in SkyWalking Cross Process Propagation Headers Protocol v2,
             * endpoint in ref could be empty, based on that, endpoint relation maybe can't be established.
             * So, I am making this source as optional.
             */
            if (endpointRelation != null) {
                sourceReceiver.receive(endpointRelation);
            }
        });

        exitSourceBuilders.forEach(exitSourceBuilder -> {
            if (nonNull(entrySpanDecorator)) {
                exitSourceBuilder.setSourceEndpointId(entrySpanDecorator.getOperationNameId());
            } else {
                exitSourceBuilder.setSourceEndpointId(Const.USER_ENDPOINT_ID);
            }
            exitSourceBuilder.setSourceEndpointName(endpointInventoryCache.get(exitSourceBuilder.getSourceEndpointId()).getName());

            exitSourceBuilder.setTimeBucket(minuteTimeBucket);
            sourceReceiver.receive(exitSourceBuilder.toServiceRelation());
            sourceReceiver.receive(exitSourceBuilder.toServiceInstanceRelation());
            if (RequestType.DATABASE.equals(exitSourceBuilder.getType())) {
                sourceReceiver.receive(exitSourceBuilder.toDatabaseAccess());
            }
        });

        // 将 slowDatabaseAccesses 集合中记录的全部DatabaseSlowStatement 对象交给 SourceReceiver 处理
        slowDatabaseAccesses.forEach(sourceReceiver::receive);
    }

    // 主要是将 TraceSegment 中的第一个 UniqueId 拼接成 String 记录到 traceId 字段中
    @Override public void parseGlobalTraceId(UniqueId uniqueId, SegmentCoreInfo segmentCoreInfo) {
        if (traceId == null) {
            StringBuilder traceIdBuilder = new StringBuilder();
            for (int i = 0; i < uniqueId.getIdPartsList().size(); i++) {
                if (i == 0) {
                    traceIdBuilder.append(uniqueId.getIdPartsList().get(i));
                } else {
                    traceIdBuilder.append(".").append(uniqueId.getIdPartsList().get(i));
                }
            }
            traceId = traceIdBuilder.toString();
        }
    }

    public static class Factory implements SpanListenerFactory {

        @Override
        public SpanListener create(ModuleManager moduleManager, TraceServiceModuleConfig config) {
            return new MultiScopesSpanListener(moduleManager, config);
        }
    }
}
